package com.project.RocketRoute.executor;

import com.alibaba.fastjson.JSONObject;
import com.project.RocketRoute.model.RocketValueEntity;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.TreeMap;
import java.util.UUID;

@Component
public class RocketMqProductExecutor {

    @Value("${rocketmq.namesrv}")
    private String namesrvAddr;
    /**
     * 消息最大大小，默认4M
     */
    @Value("${rocketmq.maxMessageSize}")
    private Integer maxMessageSize ;
    /**
     * 消息发送超时时间，默认3秒
     */
    @Value("${rocketmq.timeout}")
    private Integer sendMsgTimeout;
    /**
     * 消息发送失败重试次数，默认2次
     */
    @Value("${rocketmq.retryTimes}")
    private Integer retryTimesWhenSendFailed;

    @Value("${rocketmq.group}")
    private String groupName;

    @Value("${rocketmq.accessKey:}")
    private String accessKey;

    @Value("${rocketmq.secretKey:}")
    private String secretKey;

    private boolean showdown = true;

    private DefaultMQProducer defaultMQProducer;

    private Map<String, DefaultMQProducer> groupItems = new TreeMap<>();

//    private DefaultMQProducer getProducer() {
//        return getProducer(groupName);
//    }

    private DefaultMQProducer getProducer(String name) {
        if (StringUtils.isNotBlank(accessKey)) {
            AclClientRPCHook acl = new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
            return new DefaultMQProducer(name, acl, true, null);
        }

        return new DefaultMQProducer(name);
    }

    protected synchronized void init(){
        if (null == defaultMQProducer){
            defaultMQProducer = createProducer(groupName);
            showdown = false;
        }
    }

    private DefaultMQProducer createProducer(String groupName) {
        DefaultMQProducer producer = getProducer(groupName);
        producer.setNamesrvAddr(this.namesrvAddr);
        producer.setVipChannelEnabled(false);
        //如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
        //producer.setInstanceName(instanceName);
        producer.setMaxMessageSize(this.maxMessageSize);
        producer.setSendMsgTimeout(this.sendMsgTimeout);
        //如果发送消息失败，设置重试次数，默认为2次
        producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
        System.out.println("===============>rocket start: " + groupName);
        try {
            producer.start();
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("===============>rocket start error: " + groupName);
        }

        return producer;
    }

    public synchronized void shutdown(){
        if (null != defaultMQProducer && !showdown){
            defaultMQProducer.shutdown();
            showdown = true;
        }
    }

    public void send(String topic, String result, String... jsons){
        if (StringUtils.isEmpty(result)) {
            return;
        }

        init();
        RocketValueEntity rocketValueEntity = new RocketValueEntity();
        rocketValueEntity.setParameter(new String[jsons.length]);
        rocketValueEntity.setTaskId(UUID.randomUUID().toString());
        for(int i=0;i<jsons.length;i++){
            rocketValueEntity.getParameter()[i] = jsons[i];
        }
        String json = JSONObject.toJSONString(rocketValueEntity);
        Message sendMsg = new Message(topic, result, json.getBytes());
        try {
            defaultMQProducer.send(sendMsg);
            System.out.println("send message: " + json);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void sendGroup(String groupId, String topic, String result, String... jsons){
        if (StringUtils.isEmpty(result)) {
            return;
        }

        init();
        RocketValueEntity rocketValueEntity = new RocketValueEntity();
        rocketValueEntity.setParameter(new String[jsons.length]);
        rocketValueEntity.setTaskId(UUID.randomUUID().toString());
        for(int i=0;i<jsons.length;i++){
            rocketValueEntity.getParameter()[i] = jsons[i];
        }
        String json = JSONObject.toJSONString(rocketValueEntity);
        Message sendMsg = new Message(topic, result, json.getBytes());
        try {
//            defaultMQProducer.send(sendMsg);
            DefaultMQProducer prod = defaultMQProducer;
            if (StringUtils.isNotBlank(groupId)) {
                if (groupItems.containsKey(groupId)) {
                    prod = groupItems.get(groupId);
                } else {
                    prod = createProducer(groupId);
                    groupItems.put(groupId, prod);
                }
            }
            System.out.println(String.format("================>mq send setting groupid: %s", prod.getProducerGroup()));
            prod.send(sendMsg);
            System.out.println("send message: " + json);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
